package com.flink.streaming.web.thread;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.flink.streaming.web.common.RestResult;
import com.flink.streaming.web.service.AzKaBanSqlService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;

@Slf4j
public class AzkabanFlowLogThread implements Runnable {

    /**
     * 长度
     */
    private int length;

    /**
     * 调用次数
     */
    private int callNumber = 0;

    /**
     *
     */
    private List<String> execIds;

    /**
     * service
     */
    private AzKaBanSqlService azKaBanSqlService;

    /**
     * 连接
     */
    private WebSocketSession session;


    /**
     * 记录引用
     */
    private ConcurrentHashMap<WebSocketSession, ScheduledFuture> futureMap;

    @Override
    public void run() {

        try {
            if (callNumber > 1200) {
                doStop();
            }
            log.info("入参={}", JSONObject.toJSONString(execIds));
            Long offset = Long.valueOf(++callNumber * length);
            JSONArray array = doGetArray(execIds, offset, length);

            Map<String, JSONObject> data = azKaBanSqlService.queryFlowLog(array);
            RestResult success = RestResult.success(data);
            String resp = JSONObject.toJSONString(success);
            log.info("返回值={}", resp);
            session.sendMessage(new TextMessage(resp));

            for (Map.Entry<String, JSONObject> entry : data.entrySet()) {
                String logData = entry.getValue().getString("data");
                if (StringUtils.isEmpty(logData)) {
                    // 终止任务
                    doStop();
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
            log.error("长链接处理失败", e);
        }

    }

    private void doStop() throws IOException {
        ScheduledFuture future = futureMap.get(session);
        if (Objects.isNull(future)) {
            return;
        }
        future.cancel(false);
        session.close();
        futureMap.remove(session);
    }


    private JSONArray doGetArray(List<String> execIds, Long offset, int length) {
        JSONArray array = new JSONArray();
        for (String execId : execIds) {
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("execId", execId);
            jsonObject.put("offset", offset);
            jsonObject.put("length", length);
            array.add(jsonObject);
        }
        return array;
    }


    public AzkabanFlowLogThread(int length, List<String> execIds, AzKaBanSqlService azKaBanSqlService, WebSocketSession session, ConcurrentHashMap<WebSocketSession, ScheduledFuture> futureMap) {
        this.length = length;
        this.execIds = execIds;
        this.azKaBanSqlService = azKaBanSqlService;
        this.session = session;
        this.futureMap = futureMap;
    }

    @Override
    public boolean equals(Object o) {
        return this == o;
    }

    @Override
    public int hashCode() {
        return Objects.hash(length, callNumber, execIds, azKaBanSqlService, session, futureMap);
    }
}
